package com.zzw.utils;

import com.zzw.config.MqttConfig;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * MQTT工具类
 */
@Component
public class MqttUtil {
    private static final Logger logger = LoggerFactory.getLogger(MqttUtil.class);

    //@Autowired
    private MqttConfig mqttConfig;

    private MqttClient mqttClient;
    private MqttAsyncClient mqttAsyncClient;
    private final ConcurrentHashMap<String, IMqttMessageListener> topicListeners = new ConcurrentHashMap<>();

    /**
     * 初始化MQTT客户端
     */
    //@PostConstruct
    public void init() {
        try {
            // 创建MQTT客户端
            String clientId = mqttConfig.getClientId() + "_" + UUID.randomUUID().toString().replace("-", "");
            mqttClient = new MqttClient(mqttConfig.getBroker(), clientId, new MemoryPersistence());

            // 创建MQTT异步客户端（用于获取发布令牌）
            mqttAsyncClient = new MqttAsyncClient(mqttConfig.getBroker(), clientId + "_async", new MemoryPersistence());

            // 设置连接选项
            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(mqttConfig.getUsername());
            options.setPassword(mqttConfig.getPassword().toCharArray());
            options.setConnectionTimeout(mqttConfig.getTimeout());
            options.setKeepAliveInterval(mqttConfig.getKeepalive());
            options.setAutomaticReconnect(true);
            options.setCleanSession(true);

            // 设置回调
            mqttClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    logger.error("MQTT连接丢失: {}", cause.getMessage());

                    // 尝试自动重连
                    int retryCount = 0;
                    int maxRetries = 5;
                    boolean reconnected = false;

                    while (!reconnected && retryCount < maxRetries) {
                        try {
                            Thread.sleep(3000); // 等待3秒后重试
                            logger.info("尝试重新连接MQTT (第{}次尝试)", retryCount + 1);
                            reconnect(); // 这里会同时重连同步和异步客户端
                            if (mqttClient.isConnected() && mqttAsyncClient.isConnected()) {
                                reconnected = true;
                                logger.info("MQTT自动重连成功");
                            }
                        } catch (Exception e) {
                            logger.error("MQTT自动重连失败: {}", e.getMessage(), e);
                        }
                        retryCount++;
                    }

                    if (!reconnected) {
                        logger.error("MQTT自动重连失败，已达到最大重试次数: {}", maxRetries);
                    }
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    // 由具体的订阅处理
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    logger.debug("消息发送完成");
                }
            });

            // 为异步客户端设置回调
            mqttAsyncClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    logger.error("MQTT异步客户端连接丢失: {}", cause.getMessage());
                    // 同步客户端的connectionLost会处理重连
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    // 异步客户端不处理接收到的消息
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    try {
                        logger.debug("异步客户端消息发送完成: {}", token.getMessage().toString());
                    } catch (Exception e) {
                        logger.error("获取已发送消息信息失败: {}", e.getMessage(), e);
                    }
                }
            });

            // 连接MQTT服务器
            mqttClient.connect(options);
            mqttAsyncClient.connect(options);
            logger.info("MQTT客户端已连接到: {}", mqttConfig.getBroker());
        } catch (MqttException e) {
            logger.error("MQTT客户端初始化失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 发布消息
     *
     * @param topic   主题
     * @param payload 消息内容
     * @param qos     服务质量 (0, 1, 2)
     * @param retain  是否保留消息
     * @return 是否发送成功
     */
    public boolean publish(String topic, String payload, int qos, boolean retain) {
        try {
            if (mqttClient == null || !mqttClient.isConnected()) {
                logger.error("MQTT客户端未连接");
                return false;
            }
            MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
            message.setQos(qos);
            message.setRetained(retain);

            // 使用异步客户端发布消息以获取令牌
            IMqttDeliveryToken token = mqttAsyncClient.publish(topic, message);

            // 对于QoS > 0的消息，等待发送完成
            if (qos > 0) {
                token.waitForCompletion(mqttConfig.getTimeout());
                logger.info("消息发布完成: 主题={}, QoS={}, 保留={}", topic, qos, retain);
            }

            return true;
        } catch (MqttException e) {
            logger.error("发布消息失败: {}", e.getMessage(), e);
            return false;
        }
    }

    /**
     * 订阅主题
     *
     * @param topic    主题
     * @param qos      服务质量 (0, 1, 2)
     * @param listener 消息监听器
     * @return 是否订阅成功
     */
    public boolean subscribe(String topic, int qos, IMqttMessageListener listener) {
        try {
            if (mqttClient == null || !mqttClient.isConnected()) {
                logger.error("MQTT客户端未连接");
                return false;
            }
            mqttClient.subscribe(topic, qos, listener);
            topicListeners.put(topic, listener);
            logger.info("已订阅主题: {}", topic);
            return true;
        } catch (MqttException e) {
            logger.error("订阅主题失败: {}", e.getMessage(), e);
            return false;
        }
    }

    /**
     * 取消订阅
     *
     * @param topic 主题
     * @return 是否取消成功
     */
    public boolean unsubscribe(String topic) {
        try {
            if (mqttClient == null || !mqttClient.isConnected()) {
                logger.error("MQTT客户端未连接");
                return false;
            }
            mqttClient.unsubscribe(topic);
            topicListeners.remove(topic);
            logger.info("已取消订阅主题: {}", topic);
            return true;
        } catch (MqttException e) {
            logger.error("取消订阅失败: {}", e.getMessage(), e);
            return false;
        }
    }

    /**
     * 关闭MQTT客户端
     */
    @PreDestroy
    public void close() {
        try {
            if (mqttClient != null && mqttClient.isConnected()) {
                mqttClient.disconnect();
                mqttClient.close();
                logger.info("MQTT客户端已关闭");
            }
            if (mqttAsyncClient != null && mqttAsyncClient.isConnected()) {
                mqttAsyncClient.disconnect();
                mqttAsyncClient.close();
                logger.info("MQTT异步客户端已关闭");
            }
        } catch (MqttException e) {
            logger.error("关闭MQTT客户端失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 检查MQTT客户端是否已连接
     *
     * @return 是否已连接
     */
    public boolean isConnected() {
        return mqttClient != null && mqttClient.isConnected();
    }

    /**
     * 重新连接MQTT客户端
     *
     * @return 是否重连成功
     */
    public boolean reconnect() {
        try {
            if (mqttClient != null && !mqttClient.isConnected()) {
                mqttClient.reconnect();
                logger.info("MQTT客户端已重新连接");

                // 重新连接异步客户端
                if (mqttAsyncClient != null && !mqttAsyncClient.isConnected()) {
                    mqttAsyncClient.reconnect();
                    logger.info("MQTT异步客户端已重新连接");
                }

                // 重新订阅之前的主题
                for (String topic : topicListeners.keySet()) {
                    try {
                        IMqttMessageListener listener = topicListeners.get(topic);
                        mqttClient.subscribe(topic, 1, listener); // 默认使用QoS 1重新订阅
                        logger.info("重连后重新订阅主题: {}", topic);
                    } catch (MqttException e) {
                        logger.error("重连后重新订阅主题失败: {}, 错误: {}", topic, e.getMessage(), e);
                    }
                }

                return true;
            }
            return false;
        } catch (MqttException e) {
            logger.error("MQTT客户端重连失败: {}", e.getMessage(), e);
            return false;
        }
    }
}
